package com.lili.collection.jdbc.plugins;

import com.lili.collection.core.constans.Constans;
import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.error.OutputRunnerException;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.fields.Field;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.core.record.Record;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.RecordInfo;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.utils.NoticeUtil;
import com.lili.collection.jdbc.config.JdbcOutputConfig;
import com.lili.collection.jdbc.handler.JDBCMergeIntoHandler;
import com.lili.collection.jdbc.utils.JdbcCharUtil;
import com.lili.collection.jdbc.utils.JdbcUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.List;
import java.util.Map;

/**
 * JdbcOutputPlugin
 *
 * @author lili
 * @date 2023/2/13 20:43
 */
@Slf4j
public class JdbcOutputPlugin extends JdbcPlugin<JdbcOutputConfig> implements OutputPlugin<JdbcOutputConfig> {

    private String insertSql = null;
    private PreparedStatement preparedStatement = null;
    private Throwable tempThrowable = null;
    private int deleteOffsetNum=0;
    private volatile Connection connection;
    private boolean sourceHasValue=false;

    @Override
    public void preHandle() {
        if (!this.config.isDeleteMode()) {
            log.info("没有开启表数据清空（DELETE）模式!");
            return;
        }
        String prefixDelete = "delete from";
        StringBuilder deleteSqlBuilder = new StringBuilder(prefixDelete + Constans.SPACE_STRING);
        String schemaName = this.config.getSchemaName();
        if (StringUtils.isNotBlank(schemaName)) {
            deleteSqlBuilder.append(schemaName).append(Constans.DOT_STRING);
        }
        String tableName = this.config.getTableName();
        if (StringUtils.isBlank(tableName)) {
            return;
        }
        deleteSqlBuilder.append(tableName);
        try (Statement statement = connection.createStatement()) {
            String deleteSql = deleteSqlBuilder.toString();
            log.info("删除表数据执行,deleteSql为【{}】", deleteSql);
            statement.execute(deleteSql);
        } catch (Exception exception) {
            log.error("表【{}】删除失败", tableName, exception);
            throw new OutputRunnerException(exception);
        }
    }

    private String createMaxQuerySql() {
        String schemaName = config.getSchemaName();
        if (StringUtils.isBlank(config.getTableName())) {
            return null;
        }
        String tableName = StringUtils.isNotBlank(schemaName) ? (schemaName + "." + config.getTableName()) : (config.getTableName());
        String offsetColumn = config.getOffsetColumn();
        if (StringUtils.isBlank(offsetColumn)) {
            return null;
        }
        return "select max(" + offsetColumn + ") maxval from " + tableName;
    }

    private String createDeleteMaxValueSql() {
        String schemaName = config.getSchemaName();
        if (StringUtils.isBlank(config.getTableName())) {
            return null;
        }
        String tableName = StringUtils.isNotBlank(schemaName) ? (schemaName + "." + config.getTableName()) : (config.getTableName());
        String offsetColumn = config.getOffsetColumn();
        if (StringUtils.isBlank(offsetColumn)) {
            return null;
        }
        return "delete from " + tableName + " where " + offsetColumn + " = (select * from (select max(" + offsetColumn + ") max from " + tableName + ") tb)";
    }

    @Override
    public synchronized String queryInitialOffset() {
        ResultSet resultSet = null;
        try {
            //强制修改为手动提交
            Statement statement = connection.createStatement();
            if(config.isEnableTransaction()){
                //删除最大值
                log.info("开启事务，需要删除目标偏移字段【{}】最大值",config.getOffsetColumn());
                String deleteMaxValueSql = createDeleteMaxValueSql();
                if (StringUtils.isBlank(deleteMaxValueSql)) {
                    return null;
                }
                deleteOffsetNum = statement.executeUpdate(deleteMaxValueSql);
                log.info("表偏移量删除结果:{}", deleteOffsetNum);
            }else {
                log.info("不开启事务，不能删除目标偏移字段【{}】最大值",config.getOffsetColumn());
            }
            //查询最大值
            String maxQuerySql = this.createMaxQuerySql();
            log.info("查询偏移量SQL:{}", maxQuerySql);
            if (StringUtils.isBlank(maxQuerySql)) {
                return null;
            }
            resultSet = statement.executeQuery(maxQuerySql);
            if (resultSet.next()) {
                return resultSet.getString("maxval");
            }
        } catch (Exception e) {
            log.error("获取表的最大偏移量失败", e);
            throw new OutputRunnerException("获取表的最大偏移量失败:"+e.getLocalizedMessage());
        } finally {
            JdbcUtil.closeDb(resultSet, null, null);
        }
        return null;
    }

    @Override
    public int getDeleteOffsetNum() {
        return this.deleteOffsetNum;
    }

    @Override
    public synchronized void execute(StageContext stageContext) {
        BatchRecordContext recordContext = (BatchRecordContext) stageContext;
        RecordInfo recordInfo = new RecordInfo();
        recordContext.setIndex(this.executeNum++);
        boolean isCommit=false;
        RunnerResult<String> outputRunnerResult = this.getRunnerResult();
        boolean isRollback = false;
        List<Record> output = recordContext.getOutput();
        try {
            try {
                if(this.getTransfer()!=null && this.getTransfer().isRunError()){
                    throw new IllegalStateException("JdbcOutput异步运行错误!");
                }
                if (output.size() == 0) {
                    if((sourceHasValue || this.config.isDeleteMode()) && !connection.getAutoCommit()){
                        connection.commit();
                        log.info("-->所有数据写入完成,提交事务成功!");
                    }else {
                        log.info("-->所有数据写入完成!");
                    }
                    JdbcUtil.closeDb(null, preparedStatement, connection);
                    return;
                }
                if(this.config.isIncrementalMode()){
                    JDBCMergeIntoHandler handler=new JDBCMergeIntoHandler(connection,config);
                    handler.setTableName(JdbcUtil.getTableName(config.getSchemaName(),config.getTableName()));
                    handler.setRecordInfo(recordInfo);
                    handler.mergeIntoWithJdbc(output);
                }
                if (output.size() != 0) {
                    if (preparedStatement==null) {
                        Record firstRecord = output.get(0);
                        Map<String, Field> firstRow = firstRecord.getField().getValueAsMap();
                        String dbTypeName = connection.getMetaData().getDatabaseProductName();
                        insertSql = JdbcUtil.generateMultiInsertSql(config.getSchemaName(), config.getTableName(), firstRow.keySet(), 1,dbTypeName);
                        preparedStatement = connection.prepareStatement(insertSql);
                    }
                    for (Record record : output) {
                        Map<String, Field> row = record.getField().getValueAsMap();
                        try {
                            int paramIndex = 0;
                            for (String fieldName : row.keySet()) {
                                preparedStatement.setObject(++paramIndex, JdbcCharUtil.handleEmptyChar(row.get(fieldName).getValue()));
                            }
                            preparedStatement.addBatch();
                            recordInfo.incrementSuccessCount();
                            if(executeNum==1){
                                sourceHasValue=true;
                            }
                        } catch (Throwable throwable) {
                            recordInfo.incrementErrorCount();
                            log.error("jdbc设置值异常!", throwable);
                            throw throwable;
                        } finally {
                            recordInfo.incrementTotalCount();
                        }
                    }
                    preparedStatement.executeBatch();
                    preparedStatement.clearBatch();
                    //保存通知
                    NoticeUtil.saveNotice(config,output,"Y");
                    if (config.isIncrementalMode() && !connection.getAutoCommit()) {
                        // 只有增量需要每批次提交事务
                        connection.commit();
                        isCommit=true;
                    }else if((recordContext.getStageContextState() == StageContext.StageContextState.COMPLETE) && !connection.getAutoCommit()){
                        connection.commit();
                    }else if(connection.getAutoCommit()){
                        isCommit=true;
                    }
                    log.info("-->第{}批次数据写入完成\r\n-->写入量：{}",recordContext.getIndex(),output.size());
                }else {
                    isCommit=true;
                }
            } catch (Throwable throwable) {
                NoticeUtil.saveNotice(config,output,"N");
                if(!connection.getAutoCommit()){
                    connection.rollback();
                }
                isRollback = true;
                throw throwable;
            } finally {
                if (isRollback || recordContext.getStageContextState() == StageContext.StageContextState.COMPLETE) {
                    JdbcUtil.closeDb(null, preparedStatement, connection);
                }
            }
        } catch (Throwable throwable) {
            tempThrowable = throwable;
            RunnerMsgTransfer transfer = this.getTransfer();
            if(transfer!=null){
                transfer.setRunError(true);
                transfer.interruptRunner();
            }
            if(this.config.isEnableTransaction() && !isCommit){
                recordInfo.setErrorRecordCount(recordInfo.getSuccessRecordCount());
                recordInfo.setSuccessRecordCount(0);
            }
            recordInfo.setSuccessed(false);
            recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
            recordContext.setException(new StageRecordException(throwable));
            outputRunnerResult.setSuccessed(false);
            //错误日志收集
            ErrOutputData errOutputData = recordInfo.getErrOutputData();
            errOutputData.setBatchIndex(recordContext.getIndex());
            errOutputData.setErrorMessage(throwable.toString());
            errOutputData.setSql(insertSql);
            errOutputData.setFirstBatchError(executeNum == 1 ? true : false);
            errOutputData.setTransaction(this.config.isEnableTransaction());
            errOutputData.setType("BATCH-OUTPUT");
            log.info("-->第{}批次数据写入时发生异常\r\n-->原因：{}",
                    recordContext.getIndex(),
                    throwable.getLocalizedMessage(), throwable);
        }
        if(executeNum==1 && deleteOffsetNum>0){
            ErrOutputData errOutputData =  recordInfo.getErrOutputData();
            errOutputData.setDeleteOffsetNum(deleteOffsetNum);
            outputRunnerResult.setErrOutputData(errOutputData);
        }
        if (tempThrowable != null) {
            ErrOutputData errOutputData =  recordInfo.getErrOutputData();
            errOutputData.setDeleteOffsetNum(deleteOffsetNum);
            outputRunnerResult.setErrOutputData(errOutputData);
            if(!config.isIncrementalMode() && config.isEnableTransaction()){
                this.clearResultCount(outputRunnerResult);
            }else if(!config.isEnableTransaction()){
                this.addResultCount(recordInfo, outputRunnerResult);
            }
        } else {
            if(!config.isIncrementalMode()){
                this.addResultCount(recordInfo, outputRunnerResult);
            }
            if(config.isIncrementalMode() && isCommit){
                this.addResultCount(recordInfo, outputRunnerResult);
            }
        }
        if(this.getTransfer()!=null){
            outputRunnerResult.decrementCount(deleteOffsetNum);
        }
        recordContext.clearOutput();
    }

    private void clearResultCount(RunnerResult<String> outputRunnerResult) {
        outputRunnerResult.setSuccessRecordCount(0);
        outputRunnerResult.setTotalRecordCount(0);
        outputRunnerResult.setErrorRecordCount(0);
    }

    private void addResultCount(RecordInfo recordInfo, RunnerResult<String> outputRunnerResult) {
        outputRunnerResult.addSuccessCount(recordInfo.getSuccessRecordCount());
        outputRunnerResult.addTotalCount(recordInfo.getTotalRecordCount());
        outputRunnerResult.addErrorCount(recordInfo.getErrorRecordCount());
    }

    @Override
    public void init(JdbcOutputConfig config) {
        try {
            super.init(config);
            this.connection=this.dataSource.getConnection();
        } catch (Exception e) {
            log.error("获取连接错误!");
            throw new OutputRunnerException(e);
        }
    }

    @Override
    public void dispose() {
        JdbcUtil.closeDb(null, preparedStatement, connection);
        super.dispose();
    }
}